package com.avo.ramcache.persist.timing;

import com.avo.ramcache.config.IEntity;
import com.avo.ramcache.orm.Accessor;
import com.avo.ramcache.persist.Element;
import com.avo.ramcache.persist.EventType;
import com.avo.ramcache.persist.Listener;
import com.avo.ramcache.persist.Persister;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class TimingPersister implements Persister {

    private ConcurrentHashMap<String, Element> elements = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<Class<? extends IEntity<?>>, Listener> listeners = new ConcurrentHashMap<>();
    private boolean initialize;
    private TimingConsumer consumer;
    private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
    private volatile boolean stop;
    private final ConcurrentHashMap<String, ReentrantLock> locks = new ConcurrentHashMap<>();

    public TimingPersister() {
    }

    @Override
    public void initialize(String name, Accessor accessor, String cron) {
        if (this.initialize) {
            throw new RuntimeException("重复初始化异常");
        }
        this.elements = new ConcurrentHashMap<>();
        this.consumer = new TimingConsumer(name, cron, accessor, this);
        this.initialize = true;
    }

    @Override
    public void addListener(Class<? extends IEntity<?>> clz, Listener listener) {
        if (listener == null) {
            throw new RuntimeException("监听器为空");
        }
        this.listeners.put(clz, listener);
    }

    @Override
    public Listener getListener(Class<? extends IEntity<?>> clz) {
        return this.listeners.get(clz);
    }

    @Override
    public void put(Element element) {
        if (element == null) {
            return;
        }
        if (this.stop) {
            throw new RuntimeException("实体队列停止更新，不接受新元素");
        }
        String id = element.getIdentity();
        this.rwLock.readLock().lock();
        Lock lock = this.lockIdLock(id);
        try {
            Element prev = this.elements.get(id);
            if (prev == null) {
                this.elements.put(id, element);
                return;
            }
            EventType prevType = prev.getType();
            if (!prev.update(element)) {
                this.elements.remove(id);
            } else if (prevType == EventType.REMOVE && prev.getType() == EventType.UPDATE) {
                Listener listener = this.getListener(element.getEntityClass());
                if (listener != null) {
                    listener.notify(EventType.REMOVE, true, prev.getId(), null, null);
                }
            }
        } finally {
            this.releaseIdLock(id, lock);
            this.rwLock.readLock().unlock();
        }
    }

    @Override
    public Map<String, String> getInfo() {
        HashMap<String, String> result = new HashMap<>();
        result.put("size", Integer.toString(size()));
        result.put("state", consumer.getState().name());
        result.put("nextTime", Long.toString(consumer.getNextTime()));
        return result;
    }

    public Collection<Element> clearElements() {
        rwLock.writeLock().lock();
        try {
            ArrayList<Element> result = new ArrayList<>(this.elements.values());
            elements.clear();
            return result;
        } finally {
            rwLock.writeLock().unlock();
        }
    }

    public void saveAllNow() {
        Collection<Element> leftElements = clearElements();
        consumer.persistNow(leftElements);
    }

    public int size() {
        return elements.size();
    }

    @Override
    public void shutdown() {
        this.stop = true;
        this.consumer.stop();
        while (this.consumer.getState() != TimingConsumerState.STOPPED) {
            Thread.yield();
        }
    }

    public void flush() {
        consumer.interrupt();
    }

    private Lock lockIdLock(String id) {
        ReentrantLock lock = this.locks.get(id);
        if (lock == null) {
            lock = new ReentrantLock();
            ReentrantLock pre = this.locks.putIfAbsent(id, lock);
            lock = pre != null ? pre : lock;
        }
        lock.lock();
        return lock;
    }

    private void releaseIdLock(String id, Lock lock) {
        lock.unlock();
        this.locks.remove(id);
    }

}
